package org.porkbelly.json;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;

/**
 * <p>
 * JsonMapToTuple will convert JSON data loaded with Twitter's Elephant-bird
 * and convert it to PIG tuples. First the data is loaded with JsonLoader like
 * this:</p>
 * <p>mydata = load '/inputdata.json' USING com.twitter.elephantbird.pig.load.JsonLoader('-nestedLoad')
 *  as (json:map[]);</p>
 * <p>Once the data is loaded, it looks like "key#value" in PIG. The function
 * takes one map and returns one tuple. It expects a String for the key, since
 * all keys in PIG have to be String. The value will be coerced to Double. The
 * method gets the String representation of the value and creates a new Double.
 * </p>
 * <p>Example of JSON input data:</p>
 * <p>{"keywordCount": [{"glee": 14},{"beatles":10},{"let it be":25}] }</p>
 * <p>After JsonLoader reads the data, keywordCount has a Bag of Map entries.
 * You can pass each Map to the function and get tuples with PIG foreach. For
 * example:</p>
 * <p>keyword-data = foreach mydata generate FLATTEN(json#'keywordCount') as
 *  (keywords:map[]);</p>
 * <p>keyword = foreach keyword-data generate org.porkbelly.json.JsonMapToTuple(keywords)
 *  as (name:chararray,count:long);</p>
 * <p>The final result is you end up with rows containing tuple of (keyword,count).
 * You can then group and filter the data as you need.</p>
 * 
 * 
 * @author peter lin
 *
 */
@SuppressWarnings("rawtypes")
public class JsonMapToTuple extends EvalFunc<Tuple> {

	@Override
	public Tuple exec(Tuple input) throws IOException {
		if (input == null || input.size() <= 0) {
			return null;
		}
		HashMap map = null;
		Object v = null;
		try {
			Tuple output = TupleFactory.getInstance().newTuple(2);

			v = input.get(0);
			if (v != null) {
				if (v instanceof java.util.HashMap) {
					map = (HashMap)v;
					Iterator keyItr = map.keySet().iterator();
					while (keyItr.hasNext()) {
						String key = (String)keyItr.next();
						if (key != null && key.length() > 0) {
							Object val = map.get(key);
							val = new Double(val.toString());
							output.set(0, key);
							output.set(1,val);
						} else {
							return null;
						}
					}
				}
			}

			return output;
		} catch (ClassCastException e) {
			String msg = "Input type must be a Map. Size of input tuple: " + input.size();
			if (v != null) {
				msg += "\n - object type is: " + v.getClass().getName();
				if (v instanceof HashMap) {
					HashMap themap = (HashMap)v;
					msg += "\n - size of HashMap: " + themap.size();
					msg += "\n - details: " + themap.toString();
				}
			}
			throw new RuntimeException(msg);
		} catch (ArrayIndexOutOfBoundsException e) {
			throw new RuntimeException(
					"Function input must have even number of parameters");
		} catch (Exception e) {
			throw new RuntimeException("Error while creating a map", e);
		}
	}

    @Override
    public Schema getInputSchema(){
        return new Schema(new Schema.FieldSchema(null, DataType.BAG));
    }
    
    @Override
    public Schema outputSchema(Schema input) {
        return new Schema(new Schema.FieldSchema(null, DataType.TUPLE));
    }

}
